package edu.zju.gis.dbfg.server.controller;

import edu.zju.gis.dbfg.common.Page;
import edu.zju.gis.dbfg.common.Result;
import edu.zju.gis.dbfg.common.constant.CodeConstants;
import edu.zju.gis.dbfg.common.util.FileUtil;
import edu.zju.gis.dbfg.server.config.CommonSetting;
import edu.zju.gis.dbfg.server.constant.CodeType;
import edu.zju.gis.dbfg.server.model.Monitor;
import edu.zju.gis.dbfg.server.model.ParallelModelWithBLOBs;
import edu.zju.gis.dbfg.server.model.Role;
import edu.zju.gis.dbfg.server.model.TaskRecord;
import edu.zju.gis.dbfg.server.model.vo.PublicModelsInfo;
import edu.zju.gis.dbfg.server.model.vo.ParallelTask;
import edu.zju.gis.dbfg.server.service.MonitorService;
import edu.zju.gis.dbfg.server.service.ParallelModelService;
import edu.zju.gis.dbfg.server.service.RoleService;
import edu.zju.gis.dbfg.server.service.TaskRecordService;
import edu.zju.gis.dbfg.server.task.MonitorTasks;
import edu.zju.gis.dbfg.server.task.TaskMonitor;
import edu.zju.gis.dbfg.server.util.LogUtil;
import edu.zju.gis.dbfg.server.util.SSHHelper;
import edu.zju.gis.dbfg.server.util.XMLReader;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.util.http.fileupload.servlet.ServletFileUpload;
import org.json.JSONArray;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.multipart.MultipartHttpServletRequest;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;

/**
 * 用于处理模型计算相关业务。包括模型注册和删除，任务提交、监控、历史查看，结果下载、预览等
 *
 * @author yanlo yanlong_lee@qq.com
 * @version 1.0 2018/08/01
 */
@Slf4j
@CrossOrigin
@Controller
@RequestMapping("/parallel")
public class ParallelModelController {
    @Autowired
    private RoleService roleService;
    @Autowired
    private CommonSetting setting;
    @Autowired
    private ParallelModelService parallelModelService;
    @Autowired
    private TaskRecordService taskRecordService;
    @Autowired
    private MonitorService monitorService;
    @Autowired
    private MonitorTasks monitorTasks;
    @Value("${spring.mvc.static-path-pattern}")
    private String staticPathPattern;

    @RequestMapping(value = "/register", method = RequestMethod.POST)
    @ResponseBody
    public Result register(@SessionAttribute("userId") String userId, HttpServletRequest request) {
        Result<String> result = new Result<>();
        if (!ServletFileUpload.isMultipartContent(request)) {//判断是否为包含文件的表单数据
            result.setCode(CodeConstants.VALIDATE_ERROR)
                    .setMessage("必须有xml、jar、图片(png,jpg)三个文件被上传，或xml、py、zip（可选）、图片(png,jpg)四个文件被上传");
            return result;
        }
        List<MultipartFile> files = ((MultipartHttpServletRequest) request).getFiles("file");//获取name为file的文件（表单中name可重复）
        MultipartFile jarFile = null;//executable file, jar or py
        MultipartFile xmlFile = null;
        MultipartFile picFile = null;
        MultipartFile zipFile = null;
        CodeType codeType = CodeType.JAVA;
        for (MultipartFile file : files) {
            String filename = file.getOriginalFilename();
            if (!file.isEmpty() && filename != null) {//不处理空文件
                String ext = filename.substring(filename.lastIndexOf(".")).toLowerCase();
                switch (ext) {
                    case ".xml":
                        xmlFile = file;
                        break;
                    case ".jar":
                        jarFile = file;
                        break;
                    case ".py":
                        jarFile = file;
                        codeType = CodeType.PYTHON;
                        break;
                    case ".png":
                    case ".jpg":
                        picFile = file;
                        break;
                    case ".zip":
                    case ".egg":
                        zipFile = file;
                        break;
                }
            }
        }
        if (jarFile == null || xmlFile == null) {
            result.setCode(CodeConstants.VALIDATE_ERROR);
            result.setMessage("xml与jar（或py）文件未上传或上传失败");
            return result;
        }
        try {
            File tmpXml = new File(setting.getTemplatePath(), UUID.randomUUID().toString() + ".xml");
            xmlFile.transferTo(tmpXml);
            ParallelModelWithBLOBs model = XMLReader.readParallelModel(tmpXml);//读取并行模型模板文件
            if (model.getArtifactId() == null || model.getArtifactId().isEmpty()) {
                result.setCode(CodeConstants.VALIDATE_ERROR);
                result.setMessage("xml文件必须包含字段`artifactId`.");
                return result;
            }
            File xmlPath = new File(setting.getTemplatePath(), model.getArtifactId() + ".xml");
            if (xmlPath.exists())
                FileUtil.deletePath(xmlPath);
            FileUtil.rename(tmpXml, xmlPath);//模板文件重命名为<模型名>.xml
            model.setXmlPath(xmlPath.getAbsolutePath());
            String ext = codeType == CodeType.PYTHON ? ".py" : ".jar";
            File jarPath = new File(setting.getJarPath(), model.getArtifactId() + ext);
            if (jarPath.exists())
                FileUtil.deletePath(jarPath);
            jarFile.transferTo(jarPath);//jar文件命名为<模型名>.jar
            model.setJarPath(jarPath.getAbsolutePath());
            if (picFile != null) {
                String picName = picFile.getOriginalFilename();
                picName = picName == null ? ".png" : picName.substring(picName.lastIndexOf("."));
                File picPath = new File(setting.getPicPath(), model.getArtifactId() + picName);
                if (picPath.exists())
                    FileUtil.deletePath(picPath);
                picFile.transferTo(picPath);//pic文件命名为<模型名>.<png/jpg>
                String picUrl = staticPathPattern.replace("*", "") + picPath.getName();
                model.setPicPath(picUrl);
            }
            if (zipFile != null) {
                String zipName = zipFile.getOriginalFilename();
                zipName = zipName == null ? ".zip" : zipName.substring(zipName.lastIndexOf("."));
                File zipPath = new File(setting.getJarPath(), model.getArtifactId() + zipName);
                if (zipPath.exists())
                    FileUtil.deletePath(zipPath);
                zipFile.transferTo(zipPath);//pic文件命名为<模型名>.<png/jpg>
                model.setJarPath(model.getJarPath() + "," + zipPath.getAbsolutePath());
            }

            String paths = model.getXmlPath() + "," + model.getJarPath() + "," + model.getPicPath();
            log.info("文件上传完成，文件保存路径为:" + paths);
            result.setBody(paths);
            //TODO 优先判断模型是否  已存在
            model.setIsPublic("true");
            model.setUserId(userId);
//默认插入模型的level和department
            Role role = roleService.departmentsAndLevel(userId);
            model.setLevel(999);
            model.setDepartments(role.getDepart());

            parallelModelService.insert(model);//保存模型信息到数据库
            result.setCode(CodeConstants.SUCCESS);
        } catch (Exception e) {
            log.error("模型注册失败", e);
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("模型注册失败：" + e.getMessage());
        }
        return result;
    }

    @RequestMapping(value = "/unregister/{artifactId}", method = RequestMethod.DELETE)
    @ResponseBody
    public Result unregister(@PathVariable String artifactId) {
        try {
            ParallelModelWithBLOBs model = parallelModelService.select(artifactId);
            if (model != null) {
                FileUtil.deletePath(model.getJarPath());
                FileUtil.deletePath(model.getXmlPath());
                FileUtil.deletePath(model.getPicPath());
                parallelModelService.delete(artifactId);
                return Result.success();
            } else {
                return Result.error().setMessage("无法找到对应模型");
            }
        } catch (Exception e) {
            log.error("模型解除注册失败", e);
            return Result.error().setMessage("模型解除注册失败：" + e.getMessage());
        }
    }

    @RequestMapping(value = "/update/pic", method = RequestMethod.POST)
    @ResponseBody
    public Result updatePic(@SessionAttribute("userId") String userId, HttpServletRequest request) {
        try {
            if (!ServletFileUpload.isMultipartContent(request))//判断是否为包含文件的表单数据
                return Result.error("未选择文件");
            String artifactId = request.getParameter("artifactId");
            List<MultipartFile> files = ((MultipartHttpServletRequest) request).getFiles("file");//获取name为file的文件（表单中name可重复）
            String picUrl = "";
            if (files.get(0) != null) {
                MultipartFile picFile = files.get(0);
                String picName = picFile.getOriginalFilename();
                picName = picName == null ? ".png" : picName.substring(picName.lastIndexOf("."));
                File picPath = new File(setting.getPicPath(), artifactId + picName);
                if (picPath.exists())
                    FileUtil.deletePath(picPath);
                picFile.transferTo(picPath);//pic文件命名为<模型名>.<png/jpg>
                picUrl = staticPathPattern.replace("*", "") + picPath.getName();
            }

            ParallelModelWithBLOBs model = parallelModelService.select(artifactId);
            if (model == null)
                return Result.error("找不到artifactId为`" + artifactId + "`的模型");
            model.setPicPath(picUrl);
            parallelModelService.update(model);
            return Result.success();
        } catch (Exception e) {
            log.error("模型图片更新失败", e);
            return Result.error("模型图片更新失败：" + e.getMessage());
        }
    }

    @RequestMapping(value = "/get", method = RequestMethod.GET)
    @ResponseBody
    public Result getPublicModels(@SessionAttribute("userId") String userId,Page page) {
        Result<Page<PublicModelsInfo>> result = new Result<>();
        try {
            Page<PublicModelsInfo> models = parallelModelService.getAll(userId,page);
            result.setCode(CodeConstants.SUCCESS).setBody(models);
        } catch (Exception e) {
            log.error("模型列表读取失败", e);
            result.setCode(CodeConstants.DAO_ERROR).setMessage("模型列表读取失败：" + e.getMessage());
        }
        return result;
    }

    //根据userid获取个人数据
    @RequestMapping(value = "/getpersonal", method = RequestMethod.GET)
    @ResponseBody
    public Result getPersonalModels(@SessionAttribute("userId") String userId, Page page) {
        Result<Page<PublicModelsInfo>> result = new Result<>();
        try {
            Page<PublicModelsInfo> models = parallelModelService.selectByUserId(userId, page);
            result.setCode(CodeConstants.SUCCESS).setBody(models);
        } catch (Exception e) {
            log.error("模型列表读取失败", e);
            result.setCode(CodeConstants.DAO_ERROR).setMessage("模型列表读取失败：" + e.getMessage());
        }
        return result;
    }

    @RequestMapping(value = "/get/{artifactId}", method = RequestMethod.GET)
    @ResponseBody
    public Result getModel(@PathVariable String artifactId) {
        Result<ParallelModelWithBLOBs> result = new Result<>();
        ParallelModelWithBLOBs model = null;
        model = parallelModelService.select(artifactId);
        if (model == null)
            result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("找不到artifactId为`" + artifactId + "`的模型");
        else
            result.setCode(CodeConstants.SUCCESS).setBody(model);
        return result;
    }

    @RequestMapping(value = "/use/{artifactId}", method = RequestMethod.POST)
    @ResponseBody
    public Result postTask(@SessionAttribute("userId") String userId, @PathVariable String artifactId, @RequestBody ParallelTask requestBody) {
        Result<String> result = new Result<>();
        try {
            ParallelModelWithBLOBs model = parallelModelService.select(artifactId);
            if (model == null) {
                result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("找不到artifactId为`" + artifactId + "`的模型");
                return result;
            }
            result = postParallelModelTask(model, userId, requestBody);
        } catch (Exception e) {
            result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("数据下拉或我提交任务失败。");
        }
        return result;
    }

    //提交普通任务
    private Result<String> postParallelModelTask(ParallelModelWithBLOBs model, String userId, ParallelTask requestBody) throws IOException {
        Result<String> result = new Result<>();
        String remarks = requestBody.getJobname();
        List<String> params = requestBody.getParams();
        List<String> envSetting = requestBody.getSetting();
        String jobName = model.getArtifactId() + System.currentTimeMillis();

        JSONArray modelParam = new JSONArray(model.getParameters());

        for (int i = 0; i < params.size(); i++) {
            JSONObject obj = modelParam.getJSONObject(i);
            if (obj.has("out") || obj.has("in")) {
                String path = params.get(i);
                if (path.startsWith("ES:"))
                    params.set(i, path.replace("ES:", ""));
                else {
                    String allPath = Paths.get(setting.getUserSpaceRootPath(), userId, path).toString();
                    if (obj.has("in") && path.endsWith(".csv")) {
                        allPath = "file://" + allPath;
                    }
                    params.set(i, allPath);
                }
//                else
//                    params.set(i, Paths.get(setting.getUserSpaceRootPath(), userId, path).toString());
            }
            if (obj.has("tag")) {
                String val = params.get(i);
                String tag = obj.getString("tag");
                params.set(i, tag + " " + val);
            }
        }

        String cmd = trimChar(parallelModelService.getCmd(model, jobName, params, envSetting), '"');

        System.out.println("************************* Test CMD *************************");
        System.out.println(cmd);
        System.out.println("************************* Te END D *************************");

        if (cmd == null) {
            result.setCode(CodeConstants.SYSTEM_ERROR).setMessage("不支持的模型运行框架：" + model.getFrameworkType());
        } else
            try {
                SSHHelper.runSSH(setting.getNameNode(), setting.getUsername(), setting.getPassword(), cmd, setting.getParallelFilePath());
                log.info("a new application is submitted by " + userId + " with cmd[" + cmd + "]");
                submitTask(jobName, userId, model, params, remarks, true);
                result.setCode(CodeConstants.SUCCESS);
                result.setBody(jobName + " cmd:" + cmd);
            } catch (Exception e) {
                log.error("任务提交失败", e);
                result.setCode(CodeConstants.SYSTEM_ERROR).setMessage("任务提交失败" + e.getMessage());
            }
        return result;
    }


    @RequestMapping(value = "/kill/{jobId}", method = RequestMethod.POST)
    @ResponseBody
    public Result killJob(@PathVariable String jobId) {
        Result<String> result = new Result<>();
        try {
            Monitor monitor = monitorService.getByName(jobId);
            if (monitor == null) {
                return Result.error().setMessage("未查询到该任务：" + jobId);
            }
            String applicationId = monitor.getId();
            String cmd = setting.getYarnHome() + " application -kill " + applicationId;
            log.info("尝试kill任务:", cmd);
            SSHHelper.runSSH(setting.getNameNode(), setting.getUsername(), setting.getPassword(), cmd, setting.getParallelFilePath());
            monitor.setFinalStatus("KILLED");
            monitorService.update(monitor);
            result.setCode(CodeConstants.SUCCESS).setMessage("任务中止成功");
        } catch (IOException e) {
            log.error("任务中止失败", e);
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("任务中止失败：" + e.getMessage());
        }
        return result;
    }

    @RequestMapping(value = "/jobs", method = RequestMethod.GET)
    @ResponseBody
    public Result getJobRecords(@SessionAttribute("userId") String userId, Page page) {
        try {
            Page<TaskRecord> tasks = taskRecordService.getPagesByUser(userId, page);
            return Result.success().setBody(tasks);
        } catch (Exception e) {
            return Result.error("任务记录读取失败:" + e);
        }
    }

    @RequestMapping(value = "/jobs/{jobName}", method = RequestMethod.GET)
    @ResponseBody
    public Result getRecord(@PathVariable String jobName) {
        Result<TaskRecord> result = new Result<>();
        TaskRecord task = taskRecordService.select(jobName);
        if (task != null) {
            result.setCode(CodeConstants.SUCCESS).setBody(task);
        } else {
            result.setCode(CodeConstants.DAO_ERROR).setMessage("任务`" + jobName + "`不存在或已被删除");
        }
        return result;
    }

    @RequestMapping(value = "/jobs/{jobName}/delete", method = RequestMethod.DELETE)
    @ResponseBody
    public Result deleteRecord(@PathVariable String jobName) {
        Result result = new Result<>();
        Monitor monitor;
        monitor = monitorService.getByName(jobName);
        if (monitor == null) {
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("该记录不存在或已被删除");
        } else {
            if (monitor.getFinalStatus().equals(Monitor.FinalStatus.UNDEFINED.name())) {
                result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("该任务未完成，请先停止该任务");
            } else {
                try {
                    monitorService.delete(monitor.getId());
                    //FileOperation.deletePath(new File(setting.getJobResultPath(), monitor.getName()));//删除结果目录，存在权限问题
                    TaskRecord record = taskRecordService.select(monitor.getName());
                    if (record != null) {
                        taskRecordService.delete(record.getApplicationId());
                    }
                    result.setCode(CodeConstants.SUCCESS).setBody("success");
                } catch (Exception e) {
                    log.error("任务记录删除失败", e);
                    result.setCode(CodeConstants.SERVICE_ERROR).setMessage("任务记录删除失败：" + e.getMessage());
                }
            }
        }
        return result;
    }

    @RequestMapping(value = "/jobs/running", method = RequestMethod.GET)
    @ResponseBody
    public Result getRunningJobs(@SessionAttribute("userName") String userName) {
        Result<List<TaskRecord>> result = new Result<>();
        List<TaskRecord> tasks = taskRecordService.getByUserWithState(userName, Monitor.FinalStatus.UNDEFINED);
        result.setCode(CodeConstants.SUCCESS);
        result.setBody(tasks);
        return result;
    }

    @RequestMapping(value = "/monitor/{jobName}", method = RequestMethod.GET)
    @ResponseBody
    public Result getMonitor(@PathVariable String jobName) {
        Result<Monitor> result = new Result<>();
        Monitor monitor;
        if (jobName.startsWith("application_"))
            monitor = monitorService.select(jobName);
        else
            monitor = monitorService.getByName(jobName);
        if (monitor == null)
            result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("该任务不存在或已被删除");
        else
            result.setCode(CodeConstants.SUCCESS).setBody(monitor);
        return result;
    }

    private void submitTask(String jobName, String userId, ParallelModelWithBLOBs model, List<String> params, String tag, boolean addToMyData) {
        TaskRecord record = new TaskRecord();
        record.setApplicationId(jobName);
        record.setToolId(model.getArtifactId());
        record.setCreateUser(userId);
        record.setSubmitTime(String.valueOf(System.currentTimeMillis()));
        record.setState(Monitor.FinalStatus.UNDEFINED.name());
        record.setRemarks(tag);
        record.setOutputType("NO_SET");
        record.setParams(params.toString());
        taskRecordService.insert(record);
        monitorTasks.execute(new TaskMonitor(setting, taskRecordService, monitorService, record, model, params, addToMyData));
    }

    /***
     * 获取工作流运行状态与工作日志
     * @param jobName
     * @return
     */
    @RequestMapping(value = "/log/{jobName}", method = RequestMethod.GET)
    @ResponseBody
    public Result getjobLog(@PathVariable String jobName) {
        Result<String> result = new Result<>();
        try {
            String joblog = LogUtil.getSparkTaskLogByName(jobName, setting.getJobMonitor());
            result.setBody(joblog).setCode(CodeConstants.SUCCESS).setMessage("日志获取成功。");
        } catch (Exception e) {
            log.error("日志获取失败", e);
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("日志获取失败。");
        }
        return result;
    }

    //TODO 计算结果预览
    @RequestMapping(value = "/result/preview/{jobName}", method = RequestMethod.GET)
    @ResponseBody
    public Result previewResult(@PathVariable String jobName, Integer size, Integer ordinal) {
        return Result.error();
    }

    //TODO 计算结果上图(GeoJson)
    @RequestMapping(value = "/result/preview/{jobName}/geojson", method = RequestMethod.GET)
    @ResponseBody
    public Result previewResultAsGeoJSON(@PathVariable String jobName, int size, Integer ordinal) {
        return Result.error();
    }

    //TODO 计算结果下载
    @RequestMapping(value = "/result/download/{jobName}", method = RequestMethod.GET)
    public void downloadResult(@PathVariable String jobName, HttpServletResponse response) {
    }

    private String trimChar(String source, char un) {
        char[] chars = source.toCharArray();
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < chars.length; i++) {
            if (chars[i] != un)
                sb.append(chars[i]);
            else
                continue;
        }
        return sb.toString();
    }
}
